package com.lee.trigger;

import com.lee.TCP;
import com.lee.networkData;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.Window;

@PublicEvolving
public class SessionDataTrigger<W extends Window> extends Trigger<networkData, W> {
    private static final long serialVersionUID = 1L;
    private final long maxCount;
    private final ReducingStateDescriptor<Long> stateDesc;

    public SessionDataTrigger( ) {
        this.stateDesc = new ReducingStateDescriptor("count", new SessionDataTrigger.Sum(), LongSerializer.INSTANCE);
        this.maxCount = Long.MAX_VALUE;
    }

    public TriggerResult onElement(networkData element, long timestamp, W window, TriggerContext ctx) throws Exception {
        ReducingState<Long> count = (ReducingState)ctx.getPartitionedState(this.stateDesc);
        count.add(Long.valueOf(1L));
        TCP tcp = (TCP) element.getTransportLayer();
        if(tcp.getFIN()||tcp.getRST()) {
            if (count.get()<3){
                count.clear();
                return TriggerResult.PURGE;
            }
            count.clear();
            return TriggerResult.FIRE_AND_PURGE;
        } else {
            return TriggerResult.CONTINUE;
        }
    }

    public TriggerResult onEventTime(long time, W window, TriggerContext ctx) {
        return TriggerResult.CONTINUE;
    }

    public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception {
        ReducingState<Long> count = (ReducingState)ctx.getPartitionedState(this.stateDesc);
        if (count.get()<3){
            count.clear();
            return TriggerResult.PURGE;
        }
        return TriggerResult.FIRE_AND_PURGE;
    }

    public void clear(W window, TriggerContext ctx) throws Exception {
        ((ReducingState)ctx.getPartitionedState(this.stateDesc)).clear();
    }

    public boolean canMerge() {
        return true;
    }

    public void onMerge(W window, OnMergeContext ctx) throws Exception {
        ctx.mergePartitionedState(this.stateDesc);
    }

    public String toString() {
        return "CountTrigger(" + this.maxCount + ")";
    }



    private static class Sum implements ReduceFunction<Long> {
        private static final long serialVersionUID = 1L;

        private Sum() {
        }

        public Long reduce(Long value1, Long value2) throws Exception {
            return Long.valueOf(value1.longValue() + value2.longValue());
        }
    }
}
